package com.dahuan.producer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class PartitionProducer {

    public static void main(String[] args) {
        //TODO 1 创建kafka生产者的配置信息
        Properties props = new Properties();
        //TODO 2 指定连接kafka的集群
        //props.put( ProducerConfig. )
        props.put( "bootstrap.servers", "linux-1:9092" );
        //TODO 3 ack应答级别
        props.put( "acks", "all" );
        //TODO 4 重试次数
        props.put( "retries", 3 );
        //TODO 5 批次大小  16K  (字节)
        props.put( "batch.size", 16384 );
        //TODO 6 等待时间
        props.put( "linger.ms", 1 );
        //TODO RecordAccumulator 缓冲区大小
        props.put( "buffer.memory", 33554432 );
        //TODO k v 序列化机制
        props.put( "key.serializer", "org.apache.kafka.common.serialization.StringSerializer" );
        props.put( "value.serializer", "org.apache.kafka.common.serialization.StringSerializer" );

        //TODO 添加自定义分区
        props.put( ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.dahuan.partition.MyPartition" );

        //TODO 创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>( props );

        //TODO 发送数据
        for (int i = 0; i < 10; i++) {
            producer.send( new ProducerRecord<String, String>( "first", "dahuanhuan" + i ), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null){
                        System.out.println(metadata.partition() + "--" + metadata.offset());
                    }else {
                        exception.printStackTrace();
                    }
                }
            } );
        }
        producer.close();
    }
}
